package M

import rx.Observable
import rx.Observable.create
import rx.Subscriber
import rx.schedulers.Schedulers
import java.util.concurrent.TimeUnit

fun main(args: Array<String>) {
//    merge()
    interleavingMerge()
}

private fun merge() {

    //单层展平
    val observableObservable = create(Observable.OnSubscribe<Observable<String>> { e ->
        for (i in 8 downTo 1) {
            e.onNext(Observable.just("Item$i"))
        }
        e.onCompleted()
    })
    Observable.merge(observableObservable).subscribe(object : Subscriber<String>() {
        internal var start: Long = 0

        override fun onStart() {
            start = System.nanoTime()
        }

        override fun onNext(value: String) {
            println(value)
        }

        override fun onError(e: Throwable) {

        }

        override fun onCompleted() {
            println("time consumed:" + (System.nanoTime() - start))
        }
    })

    //三层展开
    val observablesss = create(Observable.OnSubscribe<Observable<Observable<String>>> { e ->
        for (i in 0..7) {
            e.onNext(Observable.just(Observable.just("Item== $i")))
        }
        e.onCompleted()
    })

    val merge = Observable.merge(observablesss)
    Observable.merge(merge).subscribe(object : Subscriber<String>() {
        internal var start: Long = 0

        override fun onStart() {
            start = System.nanoTime()
        }

        override fun onNext(value: String) {
            println(value)
        }

        override fun onCompleted() {
            println("time consumed:" + (System.nanoTime() - start))
        }

        override fun onError(e: Throwable) {

        }
    })


}

private fun interleavingMerge(){
    val o1 = Observable.just("1").delay(300,TimeUnit.MILLISECONDS)


    val o2 = Observable.just("2")

    Observable.merge(o1,o2).subscribe(System.out::println)

}